package com.yuan.key_code;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * WordCount类用于统计文本中每个单词出现的次数
 * 它通过Hadoop框架实现MapReduce作业
 */
public class WordCount {

    /**
     * 主函数，配置并执行MapReduce作业
     * 
     * @param args 命令行参数，包括输入文件路径和输出文件路径
     * @throws Exception 当作业配置或执行出现错误时抛出异常
     */
    public static void main(String[] args) throws Exception {
        // 创建配置对象
        Configuration conf = new Configuration();

        // 解析命令行参数，获取除Hadoop系统参数外的其他参数
        // otherArgs[0]：输入文件路径，otherArgs[1]：输出文件路径
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        // 确保至少有一个输入路径和一个输出路径
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        // 创建作业实例
        Job job = Job.getInstance(conf, "word count");
        // 设置作业的JAR包
        job.setJarByClass(WordCount.class);

        // !设置作业的Mapper类、Combiner类和Reducer类
        job.setMapperClass(WordCount.TokenizerMapper.class);
        job.setCombinerClass(WordCount.IntSumReducer.class);
        job.setReducerClass(WordCount.IntSumReducer.class);

        // 设置作业输出的键值对类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 添加输入路径
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        // 设置输出路径
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        // 等待作业完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    /**
     * !Mapper类，负责将输入的文本数据转换为<word, 1>的形式
     */
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        // 定义值1，表示每个单词出现一次
        private static final IntWritable one = new IntWritable(1);
        // 定义一个Text对象，用于存储单词
        private Text word = new Text();

        /**
         * map函数，对每个输入的文本行执行分词操作
         * 
         * @param key     输入的键，通常不使用
         * @param value   输入的值，即文本行
         * @param context 上下文对象，用于输出<word, 1>
         * @throws IOException          当写入输出时可能抛出的异常
         * @throws InterruptedException 当执行被中断时抛出的异常
         */
        public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            // 使用StringTokenizer进行分词
            StringTokenizer itr = new StringTokenizer(value.toString());
            // 遍历每个单词，输出<word, 1>
            while (itr.hasMoreTokens()) {
                this.word.set(itr.nextToken());
                context.write(this.word, one);
            }
        }
    }

    /**
     * !Reducer类，负责将每个单词的出现次数进行汇总
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        // 定义一个IntWritable对象，用于存储单词出现的总次数
        private IntWritable result = new IntWritable();

        /**
         * reduce函数，对每个单词的所有出现次数进行求和
         * 
         * @param key     输入的键，即单词
         * @param values  输入的值，即该单词出现的所有次数
         * @param context 上下文对象，用于输出<word, sum>
         * @throws IOException          当写入输出时可能抛出的异常
         * @throws InterruptedException 当执行被中断时抛出的异常
         */
        public void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            // 遍历所有出现次数，计算总和
            IntWritable val;

            // 使用迭代器遍历所有出现次数
            for (Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) {
                val = (IntWritable) i$.next();
            }
            /* !等价于下面代码
             * for(IntWritable val : values) {
             *      sum += val.get();
             * }
             */
             this.result.set(sum);
             
            // 输出<word, sum>
            context.write(key, this.result);
        }
    }
}
